package com.ecommerce.service;

import com.ecommerce.dto.QueryTypeAllProductRequestDTO;
import com.ecommerce.product.domain.IndexNameResolver;
import com.ecommerce.product.domain.Product;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.search.ClosePointInTimeRequest;
import org.elasticsearch.action.search.OpenPointInTimeRequest;
import org.elasticsearch.action.search.OpenPointInTimeResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.PointInTimeBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;

/**
 * 商品查询服务 - 使用Elasticsearch的PIT(Point-In-Time)和search_after实现深度分页查询
 */
@Slf4j
@Service
public class QueryTypeAllProductPitService {

    private final RestHighLevelClient elasticsearchClient;
    private final IndexNameResolver indexNameResolver;
    private final int defaultPageSize;
    private final TimeValue pitKeepAliveTime;

    /**
     * 构造函数注入依赖
     *
     * @param elasticsearchClient Elasticsearch高级客户端
     * @param indexNameResolver   商品索引名称解析器
     * @param defaultPageSize     默认分页大小(从配置读取)
     */
    @Autowired
    public QueryTypeAllProductPitService(
            RestHighLevelClient elasticsearchClient,
            IndexNameResolver indexNameResolver,
            @Value("${app.elasticsearch.search.default-page-size:100}") int defaultPageSize,
            @Value("${app.elasticsearch.search.pit-keep-alive:5m}") String pitKeepAlive) {
        this.elasticsearchClient = elasticsearchClient;
        this.indexNameResolver = indexNameResolver;
        this.defaultPageSize = defaultPageSize;
        this.pitKeepAliveTime = TimeValue.parseTimeValue(pitKeepAlive, TimeValue.timeValueMinutes(5), "pitKeepAlive");
    }

    /**
     * 查询符合指定的 产品大类、一级类别、二级类别 的所有产品,
     * 代码中使用PIT和search_after进行深度分页查询商品，汇总后一起返回给调用方
     *
     * @param requestDto 商品查询请求参数
     * @return 商品列表
     * @throws IOException 当Elasticsearch查询发生错误时抛出
     */
    public List<Product> queryTypeAllProduct(QueryTypeAllProductRequestDTO requestDto) throws IOException {
        log.info("Starting PIT search for products with requestDto: {}", requestDto);
        String indexName = indexNameResolver.resolveIndexName(requestDto.getProductType());

        // 1. 创建PIT(Point-In-Time)快照
        String pitId = createPointInTimeSnapshot(indexName);

        try {
            // 2. 构建基础查询条件
            BoolQueryBuilder boolQuery = buildBaseQuery(requestDto);

            // 3. 执行分页查询并收集结果
            return executePaginationSearch(pitId, boolQuery);
        } finally {
            // 4. 确保PIT资源被释放
            cleanupPointInTimeSnapshot(pitId);
        }
    }

    /**
     * 创建Point-In-Time快照
     *
     * @return PIT ID
     * @throws IOException 当创建PIT失败时抛出
     */
    private String createPointInTimeSnapshot(String indexName) throws IOException {
        OpenPointInTimeRequest pitRequest = new OpenPointInTimeRequest(indexName)
                .keepAlive(pitKeepAliveTime);

        OpenPointInTimeResponse pitResponse = elasticsearchClient.openPointInTime(pitRequest, RequestOptions.DEFAULT);
        String pitId = pitResponse.getPointInTimeId();

        log.debug("Created PIT with ID: {}", pitId);
        return pitId;
    }

    /**
     * 构建基础查询条件
     *
     * @param requestDto 商品查询请求
     * @return BoolQueryBuilder
     */
    private BoolQueryBuilder buildBaseQuery(QueryTypeAllProductRequestDTO requestDto) {
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery()
                .must(QueryBuilders.termQuery("status", 1)); // 只查询上架商品

        // 添加分类条件
        Optional.ofNullable(requestDto.getPrimaryCategory())
                .ifPresent(categoryId -> boolQuery.must(QueryBuilders.termQuery("categoryId", categoryId)));

        Optional.ofNullable(requestDto.getSecondaryCategory())
                .ifPresent(subCategoryId -> boolQuery.must(QueryBuilders.termQuery("subCategoryId", subCategoryId)));

        return boolQuery;
    }

    /**
     * 循环分页查询符合条件的产品，然后将查询到的的所有产品汇总在一起，返会给调用方
     *
     * @param pitId     PIT ID
     * @param boolQuery 查询条件
     * @return 商品列表
     * @throws IOException 当查询失败时抛出
     */
    private List<Product> executePaginationSearch(String pitId, BoolQueryBuilder boolQuery) throws IOException {
        List<Product> allProducts = new ArrayList<>();
        SearchHit[] searchHits;
        //已查询到的内容中的最后一页的排序值，在查询下一页时需要用到
        Object[] lastSortValues = null;

        do {
            // 构建搜索请求
            SearchRequest searchRequest = buildSearchRequest(pitId, boolQuery, lastSortValues);

            // 执行搜索
            SearchResponse response = elasticsearchClient.search(searchRequest, RequestOptions.DEFAULT);

            // 处理搜索结果
            searchHits = Optional.ofNullable(response)
                    .map(SearchResponse::getHits)
                    .map(SearchHits::getHits)
                    .orElse(new SearchHit[0]);

            if (searchHits.length > 0) {
                // 转换并收集结果
                List<Product> pageProducts = Arrays.stream(searchHits)
                        .map(SearchHit::getSourceAsMap)
                        .map(this::convertToProduct)
                        .collect(Collectors.toList());

                allProducts.addAll(pageProducts);

                // 获取最后一页的排序值用于下一页
                lastSortValues = searchHits[searchHits.length - 1].getSortValues();

                log.debug("Fetched {} products in current page, total so far: {}",
                        pageProducts.size(), allProducts.size());
            }

        } while (searchHits.length > 0 && searchHits.length == defaultPageSize); // 继续查询直到没有结果或不满一页

        log.info("Completed PIT search, total products found: {}", allProducts.size());
        return allProducts;
    }

    /**
     * 构建搜索请求
     *
     * @param pitId       PIT ID
     * @param boolQuery   查询条件
     * @param searchAfter 分页排序值（一般为：lastSortValues，即：已查询到的内容中的最后一页的排序值，在查询下一页时需要用到）
     * @return SearchRequest
     */
    private SearchRequest buildSearchRequest(String pitId, BoolQueryBuilder boolQuery, Object[] searchAfter) {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder()
                .query(boolQuery)
                .size(defaultPageSize)
                .sort("createTime", SortOrder.DESC) // 按创建时间降序
                .sort("_id", SortOrder.DESC)        // 确保排序唯一性
                .pointInTimeBuilder(new PointInTimeBuilder(pitId));

        // 设置分页标记
        Optional.ofNullable(searchAfter).ifPresent(sourceBuilder::searchAfter);

        return new SearchRequest().source(sourceBuilder);
    }

    /**
     * 清理PIT资源
     *
     * @param pitId PIT ID
     */
    private void cleanupPointInTimeSnapshot(String pitId) {
        if (pitId == null) return;

        try {
            ClosePointInTimeRequest closeRequest = new ClosePointInTimeRequest(pitId);
            elasticsearchClient.closePointInTime(closeRequest, RequestOptions.DEFAULT);
            log.debug("Successfully closed PIT with ID: {}", pitId);
        } catch (Exception e) {
            log.error("Failed to close PIT with ID: {}", pitId, e);
        }
    }

    /**
     * 将Map转换为Product对象
     *
     * @param sourceMap Elasticsearch返回的源数据Map
     * @return Product
     */
    private Product convertToProduct(Map<String, Object> sourceMap) {
        Product product = new Product();
        product.setId(getLongValue(sourceMap, "id"));
        product.setName(getStringValue(sourceMap, "name"));
        product.setCategoryId(getLongValue(sourceMap, "categoryId"));
        product.setSubCategoryId(getLongValue(sourceMap, "subCategoryId"));
        product.setStatus(getIntegerValue(sourceMap, "status"));
        return product;
    }

    // 辅助方法 - 安全获取Map中的值
    private Long getLongValue(Map<String, Object> map, String key) {
        return map.containsKey(key) ? ((Number) map.get(key)).longValue() : null;
    }

    private String getStringValue(Map<String, Object> map, String key) {
        return map.containsKey(key) ? (String) map.get(key) : null;
    }

    private Integer getIntegerValue(Map<String, Object> map, String key) {
        return map.containsKey(key) ? ((Number) map.get(key)).intValue() : null;
    }
}